package com.xuexi.mr.fof;

import java.io.IOException;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class RunTest {

	public static Text k=new Text();
	public static IntWritable v=new IntWritable(1);
	
	public static void main(String[] args) {
		Configuration config =new Configuration();
//		config.set("fs.defaultFS", "hdfs://node2:8020");
//		config.set("yarn.resourcemanager.hostname", "node3");
		config.set("mapred.jar", "/Users/chenbiao/Desktop/hadoop.jar");
		try {
			FileSystem fs =FileSystem.get(config);
			Job job =Job.getInstance(config);
			job.setJarByClass(RunTest.class);
			
			job.setJobName("fof");
			
			job.setMapperClass(FriendsMapper.class);
			job.setReducerClass(FriendsReducer.class);
			
			job.setMapOutputKeyClass(FoF.class);
			job.setMapOutputValueClass(IntWritable.class);
			
			job.setInputFormatClass(KeyValueTextInputFormat.class);
			
			//设置MR执行的输入文件
			FileInputFormat.addInputPath(job, new Path("/input"));
			
			//该目录表示MR执行之后的结果数据所在目录，必须不能存在
			Path outputPath=new Path("/output/fof");
			if(fs.exists(outputPath)){
				fs.delete(outputPath, true);
			}
			FileOutputFormat.setOutputPath(job, outputPath);
			
			boolean f =job.waitForCompletion(true);
			if(f){
				System.out.println("job 成功执行");
			}
			
		} catch (Exception e) {
			e.printStackTrace();
		} 
		
	}

}
